package cn.doitedu.demo4.functions;

import cn.doitedu.demo4.beans.RuleMetaBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.nio.ByteBuffer;

public class CdcMeta2RuleMetaBeanFunction implements MapFunction<String, RuleMetaBean>{
    @Override
    public RuleMetaBean map(String metaCdcJson) throws Exception {


        JSONObject jsonObject = JSON.parseObject(metaCdcJson);
        String op = jsonObject.getString("op");

        JSONObject dataObject;
        if (op.equals("d")) {
            dataObject = jsonObject.getJSONObject("before");
        } else {
            dataObject = jsonObject.getJSONObject("after");
        }

        // 取表字段
        String ruleId = dataObject.getString("rule_id");
        String modelId = dataObject.getString("model_id");
        String ruleParamJson = dataObject.getString("rule_param");
        int ruleStatus = dataObject.getIntValue("rule_status");
        String calculatorSourceCode = dataObject.getString("calculator_source_code");

        // 取出预圈选人群bitmap的序列化字节
        byte[] bytes = dataObject.getBytes("pre_select_users_bytes");
        // 将字节反序列成 bitmap 对象
        Roaring64Bitmap bitmap = Roaring64Bitmap.bitmapOf();
        bitmap.deserialize(ByteBuffer.wrap(bytes));


        Long historyEndTime = dataObject.getLong("history_end_time");

        return new RuleMetaBean(op,ruleId,modelId,calculatorSourceCode,ruleParamJson,ruleStatus,bytes,bitmap,historyEndTime);

    }
}
